package nsq

import (
	"fmt"
	"github.com/nsqio/go-nsq"
	"server/pkg/log"
)

type Handel struct {
	Retry   int64 `json:"retry"` // 消费者重试次数
	Topic   string
	Channel string
	Handel  func(message *nsq.Message) error
}

type NsqServer struct {
	Host   string `json:"host"`  // 主机
	Port   int64  `json:"port"`  // 端口
	Retry  int64  `json:"retry"` // 消费者重试次数
	Handel []*Handel
}

func (server *NsqServer) Run() {
	// 初始化消费者
	for _, handel := range server.Handel {
		server.Worker(handel)
	}
}

func (server *NsqServer) Worker(handel *Handel) {
	log.Debug("启动消费者", handel)

	// 初始化消费者
	consumer, err := nsq.NewConsumer(handel.Topic, handel.Channel, nsq.NewConfig())
	if err != nil {
		log.Error("nsq", err, fmt.Sprintf("nsq消费者初始化对象尝试创建初始化失败，重试第%d次", handel.Retry))
		return
	}

	// 设置消息处理函数
	consumer.AddHandler(nsq.HandlerFunc(handel.Handel))

	// 连接到单例nsqd
	if err = consumer.ConnectToNSQD(fmt.Sprintf("%s:%d", server.Host, server.Port)); err != nil {
		if handel.Retry <= 5 {
			handel.Retry++
			log.Error("nsq", err, fmt.Sprintf("nsq消费者连接失败，重试第%d次", handel.Retry))
			server.Worker(handel)
		} else {
			log.Error("nsq", err, "nsq消费者连接失败")
			return
		}
	}
}
